package avicit.bdp.dcs.json.service.writer;

import avicit.bdp.common.utils.database.ColumnInfo;
import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dcs.datasource.dto.JobDatasource;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.json.service.util.BuildJsonUtil;
import avicit.bdp.dcs.plugin.BaseDataxParam;
import avicit.bdp.dcs.plugin.rdbms.param.RdbmsDataxParam;
import avicit.bdp.dcs.semistructure.dto.FileTemplateDTO;
import avicit.bdp.dcs.semistructure.dto.HiveJsonDto;
import avicit.bdp.dcs.semistructure.service.TemplateFieldService;
import avicit.bdp.dcs.task.dto.ColumnMappingDto;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.tools.enums.MeasureParaTypeEnum;
import avicit.bdp.dcs.utils.DataTypeConstants;
import avicit.bdp.dcs.utils.JdbcConstants;
import com.alibaba.fastjson2.JSONObject;
import com.google.common.collect.Lists;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
public class FileWriterBuilder {

  @Autowired private TemplateFieldService templateFieldService;
  @Autowired private HiveWriterBuilder hiveWriterBuilder;
  @Autowired private IotdbWriterBuilder iotdbWriterBuilder;
  @Autowired private BuildJsonUtil buildJsonUtil;

  private static final String TYPE = "type";
  private static final String INDEX = "index";

  /** 构建文件采集Writer */
  public BaseDataxParam buildWriterByFile(
      TaskDTO taskDTO,
      JobDTO jobDTO,
      JobDatasource writerDatasource)
      throws Exception {
    // 根据列映射关系转换列名
    Map<String, String> columnMap = new HashMap<>(16);
    List<ColumnMappingDto> columnMappingDtoList =
        JSONUtils.toList(taskDTO.getMappingColumn(), ColumnMappingDto.class);
    if (CollectionUtils.isNotEmpty(columnMappingDtoList)) {
      for (ColumnMappingDto columnMappingDto : columnMappingDtoList) {
        columnMap.put(columnMappingDto.getSrcColumnName(), columnMappingDto.getDstColumnName());
      }
    }

    // 解决列之间顺序问题
    List<Map<String, Object>> columns = new ArrayList<>();
    Map<Integer, String> orderMap = new HashMap<>();
    List<ColumnInfo> columnInfoList =
        templateFieldService.getTemplateColumnListById(taskDTO.getSrcTableName());
    if (CollectionUtils.isNotEmpty(columnInfoList)) {
      int i = 0;
      for (ColumnInfo columnInfo : columnInfoList) {
        if (columnMap.get(columnInfo.getName()) == null) {
          i++;
          continue;
        }

        orderMap.put(i, columnInfo.getName());
        Map<String, Object> map = new HashMap<>();
        map.put(INDEX, String.valueOf(i));
        String type;
        if (MeasureParaTypeEnum.STRING.name().equalsIgnoreCase(columnInfo.getType())) {
          type = DataTypeConstants.STRING;
        } else if (MeasureParaTypeEnum.BOOLEAN.name().equalsIgnoreCase(columnInfo.getType())) {
          type = DataTypeConstants.BOOLEAN;
        } else if (MeasureParaTypeEnum.INTEGER.name().equalsIgnoreCase(columnInfo.getType())) {
          type = DataTypeConstants.LONG;
        } else if (MeasureParaTypeEnum.LONG.name().equalsIgnoreCase(columnInfo.getType())) {
          type = DataTypeConstants.LONG;
        } else if (MeasureParaTypeEnum.FLOAT.name().equalsIgnoreCase(columnInfo.getType())) {
          type = DataTypeConstants.DOUBLE;
        } else if (MeasureParaTypeEnum.DOUBLE.name().equalsIgnoreCase(columnInfo.getType())) {
          type = DataTypeConstants.DOUBLE;
        } else {
          type = DataTypeConstants.STRING;
        }
        map.put(TYPE, type);
        columns.add(map);
        i++;
      }
    }

    // 构建Writer
    if (JdbcConstants.IOTDB.equals(writerDatasource.getDbType())) {
      // 构建IotdbWriter
      return iotdbWriterBuilder.buildIotdbWriterBySemi(
          jobDTO, taskDTO, columnInfoList, columnMap, columnMappingDtoList, writerDatasource);
    } else if (JdbcConstants.HIVE.equals(writerDatasource.getDbType())) {
      // 构建HiveWriter
      HiveJsonDto hiveJsonDto = JSONObject.parseObject(taskDTO.getHiveJson(), HiveJsonDto.class);
      return hiveWriterBuilder.buildHiveWriterBySemi(
          jobDTO, hiveJsonDto, columnInfoList, columnMap, columnMappingDtoList, orderMap);
    } else if (JdbcConstants.GREENPLUM.equals(writerDatasource.getDbType())
        || JdbcConstants.POSTGRESQL.equals(writerDatasource.getDbType())) {
      RdbmsDataxParam dataxParam = new RdbmsDataxParam();
      dataxParam.setTableList(
          Lists.newArrayList(
              DBUtils.convertTableName(writerDatasource.getDbType(), taskDTO.getDstTableName())));
      dataxParam.setColumnList(
          buildJsonUtil.getWriterColumns(taskDTO, writerDatasource.getDbType()));
      dataxParam.setPreSql(Constants.STRING_BLANK);
      return dataxParam;
    }

    return null;
  }
}
